package cn.doitedu.rtmk.demo.press_test;

import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.lang.reflect.Type;

public class PressureTest {

    public static void main(String[] args) throws Exception {


        Configuration conf = new Configuration();
        conf.setInteger("rest.port",4040);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());


        env.setParallelism(2);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");

        DataStreamSource<String> ds = env.addSource(new TestSource());
        SingleOutputStreamOperator<EventBean> beans = ds.map(new MapFunction<String, EventBean>() {
            @Override
            public EventBean map(String value) throws Exception {

                return JSON.parseObject(value, EventBean.class);
            }
        });
        beans.keyBy(bean -> bean.getUserId())
                .process(new KeyedProcessFunction<Integer, EventBean, String>() {

                    ValueState<Long> sumState;
                    ValueState<Long> minState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sumState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("x", Long.class));
                        minState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("y", Long.class));
                    }

                    @Override
                    public void processElement(EventBean bean, KeyedProcessFunction<Integer, EventBean, String>.Context ctx, Collector<String> out) throws Exception {
                        if (bean.getEventId().equals("xxx")) {
                        } else {
                            int price = Integer.parseInt(bean.getProperties().get("item_price"));
                            sumState.update(sumState.value()==null?0:sumState.value() + price);
                            if ( (minState.value()==null? 0 : minState.value()) > price) minState.update((long) price);
                        }

                        if (RandomUtils.nextInt(1, 10) % 2 == 0) out.collect("命中了");
                    }
                }).print();

        env.execute();


    }

    public static class TestSource extends RichSourceFunction<String> {

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            /*for (int i = 0; i < 1; i++) {
                new Thread(new MyRunnable(ctx)).start();
            }*/

            while (true) {
                ctx.collect("{\"userId\":1,\"eventId\":\"item_like\",\"properties\":{\"item_price\":150},\"timeStamp\":1675935313000}");
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }

        }

        @Override
        public void cancel() {

        }
    }


    public static class MyRunnable implements Runnable {
        SourceFunction.SourceContext<String> ctx;

        public MyRunnable(SourceFunction.SourceContext<String> ctx) {
            this.ctx = ctx;
        }

        @Override
        public void run() {
            while (true) {
                ctx.collect("{\"userId\":1,\"eventId\":\"item_like\",\"properties\":{\"item_price\":150},\"timeStamp\":1675935313000}");
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

}
